package com.ams.io.network.connection;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;

import com.ams.io.BufferFactory;
import com.ams.io.network.ChannelInterestOps;

public class DatagramConnector extends NetworkConnector {
    protected static final int PACKET_BUFFER_SIZE = 1500;
    protected static HashMap<SocketAddress, DatagramConnector> connectorMap = new HashMap<SocketAddress, DatagramConnector>();;
    protected DatagramChannel channel;
    protected SocketAddress remoteAddr = null;

    private int writeDelayTime = 10; // nano second
    protected IFramer framer = null;
    protected IFramerFactory framerFactory = null;

    public DatagramConnector(DatagramChannel channel, SocketAddress remoteAddr) {
        super();
        this.channel = channel;
        this.remoteAddr = remoteAddr;
    }

    public static DatagramConnector connect(InetSocketAddress remoteAddr)
            throws IOException {
        DatagramChannel channel = DatagramChannel.open();
        channel.configureBlocking(false);
        SocketAddress bindPoint = new InetSocketAddress(0); // bind temp port;
        channel.socket().bind(bindPoint);
        DatagramConnector connector = new DatagramConnector(channel, remoteAddr);
        getDispatcher()
                .addChannelToRegister(
                        new ChannelInterestOps(channel, SelectionKey.OP_READ,
                                connector));
        connector.open();
        return connector;
    }

    public boolean finishConnect() throws IOException {
        // nothing to do
        return true;
    }

    public void readChannel() throws IOException {
        ByteBuffer readBuffer = BufferFactory.allocate(PACKET_BUFFER_SIZE);
        SocketAddress remote = null;

        if ((remote = channel.receive(readBuffer)) != null) {

            readBuffer.flip();

            DatagramConnector conn = connectorMap.get(remote);
            if (conn == null) {
                conn = new DatagramConnector(channel, remote);
                conn.setFramerFactory(framerFactory);
                for (ConnectionListener listener : listeners) {
                    conn.addListener(listener);
                }

                connectorMap.put(remote, conn);
            }
            conn.readChannel(readBuffer);
        }
    }

    protected void readChannel(ByteBuffer readBuffer) throws IOException {
        ByteBuffer[] frames = null;
        if (framer != null) {
            frames = framer.parseFrame(readBuffer);
            if (frames == null) {
                return;
            }
        } else {
            frames = new ByteBuffer[] { readBuffer };
        }
        offerInBuffers(frames);
        if (isClosed()) {
            open();
        }
        keepAlive();
    }

    public synchronized void writeToChannel(ByteBuffer[] buf)
            throws IOException {
        if (framer != null) {
            buf = framer.buildFrame(buf);
        }
        Selector writeSelector = null;
        SelectionKey writeKey = null;
        int writeTimeout = 1000;
        try {
            for (ByteBuffer frame : buf) {
                int retry = 0;
                while (channel.send(frame, remoteAddr) == 0) {
                    retry++;
                    writeDelayTime += 10;
                    // Check if there are more to be written.
                    if (writeSelector == null) {
                        writeSelector = Selector.open();
                        try {
                            writeKey = channel.register(writeSelector,
                                    SelectionKey.OP_WRITE);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                    if (writeSelector.select(writeTimeout) == 0) {
                        if (retry > 2) {
                            throw new IOException("Client disconnected");
                        }
                    }
                }
                try {
                    TimeUnit.NANOSECONDS.sleep(writeDelayTime);
                } catch (InterruptedException e) {
                }
            }
        } finally {
            if (writeKey != null) {
                writeKey.cancel();
                writeKey = null;
            }
            if (writeSelector != null) {
                writeSelector.selectNow();
            }
            keepAlive();
        }

    }

    public SocketAddress getLocalEndpoint() {
        return channel.socket().getLocalSocketAddress();
    }

    public SocketAddress getRemoteEndpoint() {
        return channel.socket().getRemoteSocketAddress();
    }

    public SelectableChannel getChannel() {
        return channel;
    }

    public void setFramerFactory(IFramerFactory framerFactory) {
        this.framerFactory = framerFactory;
        this.framer = framerFactory.create();
    }

}
